package com.lagou.utils;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;

public class HBaseUtils {

    static Configuration conf;
    static HBaseAdmin admin;
    static Connection conn;
    static {
        conf = HBaseConfiguration.create();
        conf.set(HConstants.ZOOKEEPER_QUORUM,"linux121,linux123");
        conf.set(HConstants.ZOOKEEPER_CLIENT_PORT,"2181");
        conf.setInt(HConstants.HBASE_CLIENT_OPERATION_TIMEOUT,3000);
        conf.setInt(HConstants.HBASE_CLIENT_SCANNER_TIMEOUT_PERIOD,3000);
        try {
            admin = new HBaseAdmin(conf);
            conn = ConnectionFactory.createConnection(conf);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 创建表
     */
    public static void creatTable(String tablename, String ... families) {
        try {
            HTableDescriptor htdes = new HTableDescriptor(TableName.valueOf(tablename));
            for(String f : families) {
                HColumnDescriptor family = new HColumnDescriptor(f);
                htdes.addFamily(family);
            }
            admin.createTable(htdes);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 判断表是否存在
     */
    public static boolean isTableExist(String tablename) {
        boolean isExist = false;
        try {
            isExist = admin.tableExists(Bytes.toBytes(tablename));
        } catch (IOException e) {
            e.printStackTrace();
        }
        System.out.println(isExist);
        return isExist;
    }

    /**
     * 删除表
     * @param tablename
     */
    public static void dropTable(String tablename) {
        if(isTableExist(tablename)) {
            try {
                admin.disableTable(tablename);
                admin.deleteTable(tablename);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 插入数据
     * @param tablename
     * @param family
     * @param rowkey
     * @param column
     * @param value
     */
    public static void insert(String tablename, String family, String rowkey, String column, String value) {
        try {
            Table mytable = conn.getTable(TableName.valueOf(tablename));
            byte[] rowkeyByte = Bytes.toBytes(rowkey);
            Put put = new Put(rowkeyByte);
            byte[] familyByte = Bytes.toBytes(family);
            byte[] columnByte = Bytes.toBytes(column);
            byte[] valueByte = Bytes.toBytes(value);
            put.addColumn(familyByte,columnByte, valueByte);
            put.addColumn(familyByte,Bytes.toBytes("age"),Bytes.toBytes(20));
            mytable.put(put);
            mytable.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 删除表数据
     * @param tablename 表名字
     * @param rows 根据rowkey删除
     */
    public static void delete(String tablename, String ... rows) {
        try {
            Table mytable = conn.getTable(TableName.valueOf("mytable"));
            ArrayList<Delete> deletes = new ArrayList<>();
            for(String row : rows) {
                byte[] rowKey = Bytes.toBytes(row);
                Delete delete = new Delete(rowKey);
                deletes.add(delete);
            }
            mytable.delete(deletes);
            mytable.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static void getAllRows(String tablename) {
        try {
            Table table = conn.getTable(TableName.valueOf(tablename));
            Scan scan = new Scan();
            ResultScanner scanner = table.getScanner(scan);
            for(Result result : scanner) {
                Cell[] cells = result.rawCells();
                for(Cell cell : cells) {
                    String family = Bytes.toString(CellUtil.cloneFamily(cell));
                    String rowkey = Bytes.toString(CellUtil.cloneRow(cell));
                    String column = Bytes.toString(CellUtil.cloneQualifier(cell));
                    String value = Bytes.toString(CellUtil.cloneValue(cell));
                    System.out.println("family:" + family + "...rowkey:" + rowkey + "...column:" + column + "...value:" + value);
                }
            }
            table.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static void getRow(String tablename, String row) {
        try {
            Table table = conn.getTable(TableName.valueOf(tablename));
            byte[] rowkeyByte = Bytes.toBytes(row);
            Get get = new Get(rowkeyByte);
            Result result = table.get(get);
            for(Cell cell : result.rawCells()) {
                String family = Bytes.toString(CellUtil.cloneFamily(cell));
                String rowkey = Bytes.toString(CellUtil.cloneRow(cell));
                String column = Bytes.toString(CellUtil.cloneQualifier(cell));
                String value = Bytes.toString(CellUtil.cloneValue(cell));
                long timestamp = cell.getTimestamp();
                SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                String time = sdf.format(timestamp);
                System.out.println("family:" + family + "...rowkey:" + rowkey + "...column:" + column + "...value:" + value + "...time:" + time);
            }
            table.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {
//        HBaseUtils.creatTable("mytable","f1");
//        isTableExist("mytable");
//        dropTable("mytable");
        insert("mytable","f1","3","name","jack");
//        delete("mytable","1");
//        getAllRows("mytable");
//        getRow("mytable","2");

    }
}
